import com.mongodb.BasicDBObject
import com.mongodb.casbah.Imports.{MongoClient, MongoClientURI}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
object SourceMongodb {
  case class User (_id: String, name: String, age: String, sex:String)
  def main(args: Array[String]): Unit = {
    // 1、获取流式环境变量
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 2、添加自定义source  连接mongodb
    val dataStream = env.addSource(new MongodbSource)

    val tenv= StreamTableEnvironment.create(env)

    while (true) {
      val table = tenv.fromDataStream(dataStream)
      // 3、打印数据
      tenv.toRetractStream[User](table).print()
      // 4、执行任务
      env.execute("MongoDB Job")
    }
  }

  //自定义source连接mongodb
  class MongodbSource extends RichSourceFunction[User] {
    val uri = "mongodb://116.63.47.53:18012"
    var mongoClient:MongoClient = _
    override def open(parameters: Configuration): Unit = {
      //连接本地mongodb
      mongoClient = MongoClient(MongoClientURI(uri))
    }

    override def run(ctx: SourceFunction.SourceContext[User]): Unit = {
      //TODO 获取数据库
      val database = mongoClient("test")
      //TODO 获取集合
      val coll = database("test")
      //TODO 取出数据
      val query = new BasicDBObject("name", "张三")
      val cursorType = coll.find(query)
      while (true) {
        if (cursorType.nonEmpty) {
          val oneData = cursorType.next() //拿出一条数据
          ctx.collect(
            User(
              _id = oneData.get("_id").toString,
              name = oneData.get("name").toString,
              age = oneData.get("age").toString,
              sex = oneData.get("sex").toString,
            )
          )
        }
      }
    }

    override def cancel(): Unit = {
      if (mongoClient != null) mongoClient.close
    }
  }
}

